package com.shujia.core

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCount {
  /**
   * 将Spark程序提交到集群上运行
   * 1、注释掉 conf.setMaster("local")
   * 2、修改输入输出路径为HDFS上的路径
   * 3、将代码打成jar包上传至虚拟机
   * 4、使用spark-submit提交任务
   * spark-submit --class com.shujia.Demo1WordCount --master yarn-client --executor-memory 512m --num-executors 2 --executor-cores 1 Spark-1.0-jar-with-dependencies.jar
   * spark-submit --class com.shujia.Demo1WordCount --master yarn-cluster --executor-memory 512m --num-executors 2 --executor-cores 1 Spark-1.0-jar-with-dependencies.jar
   *
   *
   */
  def main(args: Array[String]): Unit = {
    // 创建Spark的入口

    // SparkConf对象
    val conf: SparkConf = new SparkConf()

    // 给Spark程序赋予一个名字，如果提交到Yarn上 可以再8088中看到
    conf.setAppName("Demo1WordCount")

    // 指定Spark的运行方式为local方式，即在本地运行
    /**
     * local[1] 一个线程
     * local[4] 四个线程
     * local[*] CPU所允许的最大线程数
     */
    conf.setMaster("local")

    // 创建SparkContext上下文环境
    val sc: SparkContext = new SparkContext(conf)

    // 1、读取文件
    /**
     * RDD:弹性分布式数据集
     * 在这里可以把它当成Scala中的集合来使用
     * 实际上RDD是编写Spark代码时的通用数据模型
     *
     * textFile方法 ：虽然它是Spark提供的方法，但底层还是调用了MapReduce读取HDFS文件那一套东西
     * 实际上还是使用了TextInputFormat读取数据并进行格式化
     * 切片规则也是MR那一套，遵循着FileInputFormat类中的getSplits方法，默认每个切片大小也是128M
     */
    val linesRDD: RDD[String] = sc.textFile("Spark/data/words/")

    // 2、将每个单词提取出来
    // flatMap在Spark中称之为 算子
    // 大部分 算子 在执行完之后 返回的还是RDD
    val wordsRDD: RDD[String] = linesRDD.flatMap(line => line.split(","))

    // 3、对每个单词进行分组
    val groupRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)

    // 4、统计每个单词的数量
    val wordCount: RDD[String] = groupRDD.map(kv => kv._1 + "," + kv._2.size)

    //    // 5、打印
    //    wordCount.foreach(println)

    val cf: Configuration = new Configuration()

    val fs: FileSystem = FileSystem.get(cf)
    val path: Path = new Path("Spark/data/wordCnt")
    // 判断输出路径是否存在 存在则删除
    if (fs.exists(path)) {
      fs.delete(path, true)
    }

    // 将结果保存到文件
    wordCount
      .saveAsTextFile("Spark/data/wordCnt")

    while (true) {

    }

  }

}
